package com.dw.receiver.strategy.impl;

import com.dw.common.bean.ResourceBean;
import com.dw.receiver.ReceiverConfiguration;
import com.dw.receiver.strategy.ResourceSplit;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class ResourceSplitImpl implements ResourceSplit {
    private final ResourceBean resource;
    private final int splitCount;

    public ResourceSplitImpl(ResourceBean resource, int splitCount) {
        this.resource = resource;
        this.splitCount = Math.min(splitCount, ReceiverConfiguration.MAX_NODE_COUNT);
    }

    @Override
    public List<ResourceBean> split() {
        List<ResourceBean> result = new CopyOnWriteArrayList<>();

        int avgSize = Math.toIntExact(resource.getLength() / splitCount);
        long rest = resource.getLength();
        long offset = 0;
        while (rest > 0) {
            ResourceBean bean = new ResourceBean();
            bean.setOwner(resource.getOwner());
            bean.setName(resource.getName());
            bean.setOffset(offset);
            if (rest > avgSize * 2L) {
                bean.setLength(avgSize);
                offset += avgSize;
                rest -= avgSize;
            } else {
                bean.setLength(rest);
                rest = 0;
            }
            result.add(bean);
        }
        return result;
    }
}
